﻿using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace DataService.Helper.Kafka
{
    public abstract class KafkaBase
    {
        /// <summary>
        /// 获取Kafka服务器地址
        /// </summary>
        /// <param name="brokerNameKey">配置文件中Broker服务器地址的key的名称</param>
        /// <returns>返回获取到的Kafka服务器的地址明细</returns>
        public string GetKafkaBroker(string brokerNameKey = "Broker")
        {
            string kafkaBroker = System.Configuration.ConfigurationManager.AppSettings[brokerNameKey];
            if (string.IsNullOrEmpty(kafkaBroker) || string.IsNullOrWhiteSpace(kafkaBroker) || kafkaBroker.Length <= 0)
            {
                throw new ArgumentNullException("Kafka消息服务器地址不能为空！");
            }
            return kafkaBroker;
        }

        /// <summary>
        /// 主题名称
        /// </summary>
        /// <param name="topicNameKey">配置文件中主题的key名称</param>
        /// <returns>返回获取到的主题的具体值</returns>
        public string GetTopicName(string topicNameKey = "Topic")
        {
            string topicName = System.Configuration.ConfigurationManager.AppSettings[topicNameKey];

            if (string.IsNullOrEmpty(topicName) || string.IsNullOrWhiteSpace(topicName) || topicName.Length <= 0)
            {
                throw new ArgumentNullException("消息所属的主题不能为空！");
            }
            return topicName;
        }
        /// <summary>
        /// 组
        /// </summary>
        /// <returns></returns>
        public string GetGroupID(string groupIDKey = "GroupID")
        {
            string groupID = System.Configuration.ConfigurationManager.AppSettings[groupIDKey];
            if (string.IsNullOrEmpty(groupID) || string.IsNullOrWhiteSpace(groupID) || groupID.Length <= 0)
            {
                throw new ArgumentNullException("消息所属的组不能为空！");
            }
            return groupID;
        }
        /// <summary>
        /// 分区
        /// </summary>
        /// <returns></returns>
        public List<int> GetPartition(string partitionsKey = "Partitions")
        {

            List<int> partitions = new List<int>();
            System.Configuration.ConfigurationManager.AppSettings[partitionsKey].Split(',').ToList().ForEach(x =>
            {
                partitions.Add(Convert.ToInt32(x));
            });
            return partitions;
        }
        /// <summary>
        /// 一次消费消息数量
        /// </summary>
        /// <returns></returns>
        public int GetConsumerCount(string consumerCountKey = "ConsumerCount")
        {
            int count = Convert.ToInt32(System.Configuration.ConfigurationManager.AppSettings[consumerCountKey]);
            return count;
        }

        /// <summary>
        ///  写日志
        /// </summary>
        /// <param name="type">consumer,producer</param>
        /// <param name="info"></param>
        /// <param name="args"></param>
        public static void WriteLog(string type, string info, params object[] args)
        {
            try
            {
                string filelog = string.Format(@"{0}\{1}\", AppDomain.CurrentDomain.BaseDirectory, "kafkaLog");
                if (!Directory.Exists(filelog))
                {
                    Directory.CreateDirectory(filelog);
                }
                filelog = string.Format(@"{0}\{1}\{2}{3}", AppDomain.CurrentDomain.BaseDirectory, "kafkaLog", type + "_" + DateTime.Now.ToString("yyyy-MM-dd"), ".log");
                using (StreamWriter sw = File.AppendText(filelog))
                {
                    sw.WriteLine(DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss 信息：\r\n"));
                    sw.WriteLine(string.Format(info, args));
                    sw.WriteLine("\r\n");
                    sw.Flush();
                    sw.Close();
                }
            }
            catch (Exception ex)
            {
                throw ex;
            }
        }
    }

}
